package com.audience.testtractor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;

public class SelectorThread extends ThreadLocal<LinkedBlockingDeque> implements Runnable{
    // 每个线程对应一个selector，
    // 多线程情况下，该主机，该程序的并发客户端被分配到多个selector上
    // 注意，每个客户端，只绑定到其中一个selector
    // 其实不会有交互问题

    Selector selector = null;
//    LinkedBlockingDeque<Channel> lbq = new LinkedBlockingDeque<>();
    LinkedBlockingDeque<Channel> lbq=get();  //lbq 在接口或者类中是固定使用方式逻辑写死了，你需要是lbq每个线程持有自己的独立对象
    SelectorThreadGroup stg;

    @Override
    protected LinkedBlockingDeque<Channel> initialValue(){
        return new LinkedBlockingDeque<>();  //你要丰富的是这里！ pool...
    }

    SelectorThread(SelectorThreadGroup stg){
        try {
            this.stg=stg;
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {

        //Loop
        while (true){
            try {
                //1,select()
//                System.out.println(Thread.currentThread().getName()+"    : before select......"+ selector.keys().size());
                int nums = selector.select();  //阻塞  wakeup()
//                System.out.println(Thread.currentThread().getName()+"    : after select......"+ selector.keys().size());
                //2,处理selectkeys
                if (nums>0){
                    Set<SelectionKey> keys=selector.selectedKeys();
                    Iterator<SelectionKey> iter=keys.iterator();
                    while (iter.hasNext()){  //线性处理的过程
                        SelectionKey key=iter.next();
                        iter.remove();
                        if (key.isAcceptable()){  //复杂，接受客户端的过程（接受之后，要注册，多线程下，新的客户端，注册到哪里呢？）
                            acceptHandler(key);
                        } else if (key.isReadable()){
                            readHandler(key);
                        } else if (key.isWritable()){

                        }
                    }
                }
                //3,处理一些task : listen client
                if (!lbq.isEmpty()){
                    Channel c=lbq.take();
                    if (c instanceof  ServerSocketChannel){
                        var server=(ServerSocketChannel) c;
                        server.register(selector,SelectionKey.OP_ACCEPT);
                        System.out.println(Thread.currentThread().getName()+" register listen");
                    }else if (c instanceof SocketChannel){
                        var client=(SocketChannel) c;
                        var buffer=ByteBuffer.allocateDirect(4096);
                        client.register(selector,SelectionKey.OP_READ,buffer);
                        System.out.println(Thread.currentThread().getName()+" register client: "+client);
                    }
                }
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void readHandler(SelectionKey key) {
        System.out.println(Thread.currentThread().getName()+" read...");
        ByteBuffer buffer = (ByteBuffer)key.attachment();
        SocketChannel client = (SocketChannel)key.channel();
        buffer.clear();
        while (true){
            try {
                int num=client.read(buffer);
                if (num>0){
                    buffer.flip(); //将读到的内容翻转，然后直接写出
                    while (buffer.hasRemaining()){
                        client.write(buffer);
                    }
                    buffer.clear();
                }else if (num==0){
                    break;
                }else if (num<0){
                    //客户端断开了
                    System.out.println("client: "+client.getRemoteAddress()+"closed......");
                    key.cancel();
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void acceptHandler(SelectionKey key) {
        System.out.println(Thread.currentThread().getName()+" acceptHandler...");
        ServerSocketChannel server =  (ServerSocketChannel)key.channel();
        try {
            SocketChannel client=server.accept();
            client.configureBlocking(false);

            //choose a selector and register!!
            stg.nextSelectorV3(client);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void setWorker(SelectorThreadGroup stgWorker) {
        this.stg = stgWorker;
    }
}
